package com.rrd.dw.mr.gzip.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.euler.commons.web.JsonMapperBuilder;
import com.rrd.dw.utils.HadoopUtils;
import com.ucredit.babel.common.kafka.PostRecord;
import com.ucredit.babel.common.model.BaseModel;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Pattern;

/**
 * @author xj
 * @version 回溯三方数据，修复遗漏数据
 */
public class TXCreditRecallMR extends Configured implements Tool {
    public static final String CRT_FLAG = "\001";
    public static final Pattern CRT_PATTERN = Pattern.compile(CRT_FLAG);
    private static final List<ParseBase> ParserList = getParseList();

    private static List<ParseBase> getParseList() {
        ArrayList<ParseBase> parsers = new ArrayList<>();
        parsers.add(new TxCreditMainParse());
        parsers.add(new TxCreditSiFaParse());
        parsers.add(new TxCreditShiXinParse());
        parsers.add(new TxCreditFengXianParse());
        parsers.add(new TxCreditGuQuanParse());
        return parsers;
    }


    private static Map<String, String> getOutPath(String dt) {
        Map<String, String> map = new HashMap<String, String>();
        for (ParseBase parser : ParserList) {
            map.put(parser.getType() + parser.getSubType(), "/data/fdm/" + parser.getTableName() + "/dt=" + dt + "/");
        }
        return map;
    }

    public static class CallMap extends Mapper<LongWritable, Text, Text, Text> {
        private Map<String, List<ParseBase>> parserListMap = new HashMap<>();

        @Override
        protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            setParserMap();
        }

        private void setParserMap() {
            for (ParseBase parser : getParseList()) {
                List<ParseBase> parsers;
                if (parserListMap.containsKey(parser.getType())) {
                    parsers = parserListMap.get(parser.getType());
                } else {
                    parsers = new ArrayList<>();
                }
                parsers.add(parser);
                parserListMap.put(parser.getType(), parsers);
            }
        }


        public static long dateToStamp(String s, String pattern) {
            try {
                SimpleDateFormat simpleDateFormat = new SimpleDateFormat(pattern);
                Date date = simpleDateFormat.parse(s);
                return date.getTime();
            } catch (ParseException e) {
                return new Date().getTime();
            }
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            String[] tmparr = CRT_PATTERN.split(value.toString());
            JSONObject jsonObj = JSON.parseObject(tmparr[2]);
            JSONObject job = new JSONObject();
            JSONObject data = jsonObj.getJSONObject("reportJson") == null ? jsonObj : jsonObj.getJSONObject("reportJson");
            job.put("jobID", tmparr[0]);
            job.put("finished", true);
            job.put("requested", true);
            job.put("successful", true);
            job.put("startTime", "");
            job.put("endTime", "");
            job.put("dataFile", null);
            job.put("result", null);
            job.put("systemID", "HAO_HUAN");
            job.put("query", "");
            job.put("tryTimes", 1);
            job.put("extra", null);
            JSONObject record = new JSONObject();
            record.put("timestamp", dateToStamp(tmparr[3], "yyyy-MM-dd HH:mm:ss"));
            record.put("timeInCache", "");
            record.put("userKey", tmparr[1]);
            record.put("loanKey", "");
            record.put("type", "TX_CREDIT_REPORT_RECORD");
            record.put("data", data);
            record.put("idNumber", "");
            record.put("job", job);
            JSONObject input = new JSONObject();
            input.put("type", "TX_CREDIT_REPORT_RECORD");
            input.put("dataJson", record.toJSONString());
            input.put("action", "PUT");
            input.put("resendTimes", 0);
            input.put("md5", "");

            List<ParseBase> parsers = parserListMap.get(input.getString("type"));
            if (parsers != null && parsers.size() > 0) {
                String i = input.toJSONString();
                System.out.println(i);
                PostRecord r = JsonMapperBuilder.build().readValue(i, PostRecord.class);
                BaseModel<?> model = (BaseModel<?>) r.getRecord();
                if (r.getAction().equals("PUT")) {
                    for (ParseBase parser : parsers) {
                        parseValue(model, parser, context);
                    }
                }
            }
        }

        protected void parseValue(BaseModel<?> model, ParseBase parser, Context context) throws IOException, InterruptedException {
            String successful = "1";
            if (model != null) {
                if (parser.parseJob(context, model, successful)) {
                    context.getCounter("counter", "user." + parser.getType() + parser.getSubType()).increment(1);
                } else {
                    context.getCounter("counter", "user." + parser.getType() + parser.getSubType() + ".false").increment(1);
                }
            } else {
                context.getCounter("error", "user." + parser.getType() + parser.getSubType() + ".null").increment(1);
            }
        }
    }

    public static class RecallReduce extends Reducer<Text, Text, Text, Text> {
        private MultipleOutputs<Text, Text> mos;
        private NullWritable key2 = NullWritable.get();
        private String dt = null;
        private Map<String, String> map = null;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            this.mos = new MultipleOutputs<Text, Text>(context);
            Configuration conf = context.getConfiguration();
            this.dt = conf.get("dt");
            this.map = getOutPath(dt);
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            this.mos.close();
        }

        private String keystr = null;

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            keystr = key.toString();
            System.out.println("REDUCE KEY:" + keystr + "|" + this.map.get(keystr));
            for (Text val : values) {
                context.getCounter("reduce", "reduce.success").increment(1);
                this.mos.write("textinfo", key2, val, this.map.get(keystr));
            }

        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs", "org.apache.hadoop.io.compress.GzipCodec");
        Job job = Job.getInstance(conf);
        String dt = conf.get("dt");
        String i = conf.get("i", "0");
        String output = "/tmp/bdm/recall_third_party" + i;
        String inputDir = "/data/dwtmp/tx_credit_20201001_20201207/dt=" + dt;
        System.out.println("param.input.path=" + inputDir);
        System.out.println("param.out.path=" + output);
        Map<String, String> outputMap = getOutPath(dt);
        for (String outputPath: outputMap.values()){
            HadoopUtils.delete(conf, new Path(outputPath));
        }

        Path outpath = new Path(output);
        HadoopUtils.delete(conf, outpath);
        TextInputFormat.setInputPaths(job, new Path(inputDir));
        TextOutputFormat.setOutputPath(job, outpath);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        job.setJarByClass(TXCreditRecallMR.class);
        job.setMapperClass(TXCreditRecallMR.CallMap.class);
        job.setReducerClass(TXCreditRecallMR.RecallReduce.class);
        MultipleOutputs.addNamedOutput(job, "textinfo", TextOutputFormat.class, Text.class, Text.class);
        int stat = job.waitForCompletion(true) ? 0 : 1;
        for (ParseBase parser : ParserList) {
            System.out.println(parser.getComment() + parser.getSubType() + ":" + job.getCounters().getGroup("counter").findCounter("user." + parser.getType() + parser.getSubType()).getValue());
            System.out.println(parser.getComment() + parser.getSubType() + "查询is Null次数：" + job.getCounters().getGroup("error").findCounter("user." + parser.getType() + parser.getSubType() + ".null").getValue());
        }
        System.out.println("異常次數：" + job.getCounters().getGroup("error").findCounter("ClassCastException").getValue());
        System.out.println("reduce.null次数：" + job.getCounters().getGroup("reduce").findCounter("reduce.null").getValue());
        System.out.println("reduce.success次数：" + job.getCounters().getGroup("reduce").findCounter("reduce.success").getValue());
        return stat;
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new TXCreditRecallMR(), args);
        System.exit(res);
    }

}
